/**
 * Alipay.com Inc.
 * Copyright (c) 2004-2022 All Rights Reserved.
 */
package com.example.spider.proxy.core.repository;

import cn.hutool.core.net.NetUtil;
import cn.hutool.core.util.StrUtil;
import com.example.spider.config.SpiderMetrics;
import com.example.spider.config.ThreadPoolConfiguration;
import com.example.spider.proxy.core.ProxyInfo;
import com.example.spider.proxy.core.ProxyRepository;
import com.example.spider.proxy.core.ProxyTester;
import com.example.spider.proxy.core.distinct.IpPortDistincter;
import com.example.spider.web.ProgressController;
import jakarta.annotation.Nonnull;
import lombok.extern.slf4j.Slf4j;
import org.shoulder.batch.progress.BatchProgress;
import org.shoulder.batch.progress.BatchProgressCache;
import org.shoulder.batch.progress.BatchProgressRecord;
import org.shoulder.core.concurrent.Threads;
import org.shoulder.core.exception.CommonErrorCodeEnum;
import org.shoulder.core.util.AddressUtils;
import org.shoulder.core.util.AssertUtils;
import org.shoulder.core.util.ContextUtils;
import org.shoulder.core.util.StringUtils;
import org.shoulder.monitor.concurrent.MonitorableThreadPool;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

/**
 * ipproxy 存储
 *
 * @author lym
 */
@Slf4j
public abstract class AbstractProxyRepository implements ProxyRepository, BeanNameAware {

    protected String name;

    @Autowired
    @Qualifier(ThreadPoolConfiguration.IP_PROXY_TESTER)
    private MonitorableThreadPool ipTestExecutorService;

    @Lazy
    @Autowired
    protected Map<String, ProxyRepository> proxyRepositoryMap;

    @Autowired
    @Qualifier("spiderMetrics")
    private SpiderMetrics metrics;

    @Lazy
    @Autowired
    protected Map<String, ProxyTester> proxyTesterMap;

    @Autowired
    protected IpPortDistincter          ipPortDistincter;

    @Autowired
    private BatchProgressCache progressCache;

    private final ReentrantLock lock = new ReentrantLock();

    protected static String calculateKey(ProxyInfo proxyInfo) {
        return proxyInfo.getIp() + ":" + proxyInfo.getPort();
    }

    public void setBeanName(@Nonnull String beanName) {
        this.name = beanName;
    }

    private static boolean canSkipTest(String name, ProxyInfo proxyInfo) {
        LocalDateTime lastVerifyTime = proxyInfo.getLastVerificationTime();
        if (lastVerifyTime == null) {
            // 第一次测
            return false;
        }
        Duration lastValidatePassed = Duration.between(LocalDateTime.now(), lastVerifyTime);
        // 默认最小测试间隔时间（保护代理池和测试资源池），实际还受扫描间隔影响
        Duration validateDuration = Duration.ofMinutes(1);
        if (name.equals(SOURCE) && proxyInfo.getTotalCount() > 5) {
            // 前三次间隔10分钟，以后间隔1小时
            validateDuration = Duration.ofMinutes(60);

        }
        return validateDuration.compareTo(lastValidatePassed) < 0;
    }

    // source测试：用http测一下，ok就晋级，不行再用 socks测一下http，能用就晋级，连续x次不可用拉黑。测试频率：0min、10min、1h、2h、3h... 最多24次

    // usable测试：
    // 信息不全：测试出类型，用socks测一下，用http测一下，测一下访问 https，补全信息，信息完整、可用率达标就放入stable。不可用太多次，重置请求历史，放回source
    // 信息全：访问一下http，看成功率决定是否晋级 / 淘汰

    // stable测试：访问一下http，ok就保留，成功率低于一定程度，重置请求历史放到 usable

    /**
     * 定期调度，有 @Async 必须暴露给接口
     */
    @Async
    @Scheduled(cron = "0 */1 * * * ?")
    @Override
    public void validateAllProxy() {
        boolean locked = lock.tryLock();
        if(!locked) {
            return;
        }
        try {
            validateAllProxy0();
        } finally {
            lock.unlock();
        }
    }

    public void validateAllProxy0() {
        int totalSize = size();
        metrics.repoNum(name).set(totalSize);

        String key = name.contains("SOURCE") ? "source" : StrUtil.containsIgnoreCase(name, "usable") ? "usable" : name.contains("STABLE") ? "stable" : "x";
        String settings = ContextUtils.getProperty("spider.proxy.validate." + key + ".enable");
        boolean disable = StrUtil.containsAnyIgnoreCase(settings, "n", "f");
        if (disable) {
            return;
        }

        // 1. load
        Iterator<ProxyInfo> it = getIterator();
        int scanNum = 0;
        AtomicInteger promotedNum = new AtomicInteger();
        AtomicInteger eliminateNum = new AtomicInteger();
        log.trace("START {}.validateAllProxy ", getClass().getSimpleName());

        String taskId = "validateAllProxy_" + name;
        BatchProgressRecord lastRecord = progressCache.getProgress(taskId).getBatchProgress();
        if (lastRecord != null) {
            boolean finished = lastRecord.hasFinish();
            if (!finished) {
                return;
            }
            long min = Duration.between(lastRecord.getStopTime(), LocalDateTime.now()).toMinutes();
            boolean canRun = switch (name) {
                case SOURCE -> min > 10;
                case USABLE -> min > 5;
                default -> min > 3;
            };
            if (!canRun) {
                return;
            }
        }

        BatchProgress progress = new BatchProgress();
        progress.setId(taskId);
        progress.setTotal(totalSize);
        progress.setOnFinishCallback((id, p) ->
                log.info("END {}.validateAllProxy(), cost={}, scanNum={}, passNum={}, noPassNum={}, promotedNum={}, eliminateNum={}.",
                        name, Duration.ofMillis(progress.calculateProcessedTime()), progress.getTotal(), progress.getSuccessNum(), progress.getFailNum(), promotedNum, eliminateNum));
        progress.start();
        progressCache.triggerFlushProgress(progress);
        log.warn("validateAllProxy({}) 开始校验，进度： {} ", name, ProgressController.getProgressUrl(progress.getId()));


        while (it.hasNext()) {
            scanNum++;

            ProxyInfo proxyInfo = it.next();
            // 重置评分
            //proxyInfo.cleanHistory();

            ipTestExecutorService.execute(() -> {
                if (removeInvalid(proxyInfo)) {
                    progress.addFail(1);
                    return;
                }

                // 短时间校验过的暂时无需校验（实现逻辑：计算差时、redis去重，延时消息）
                if (canSkipTest(name, proxyInfo)) {
                    progress.addSuccess(1);
                    return;
                }
                // 测试、暂不需要测试：source：

                // 2. validate
                boolean validatePass = validateAndUpdateProxyInfo(proxyInfo);
                if (validatePass) {
                    progress.addSuccess(1);
                } else {
                    progress.addFail(1);
                }
                // 观察
                String op = "none";
                // 3. processResult
                if (ableToPromoted(proxyInfo)) {
                    // 晋升
                    doPromoted(proxyInfo);
                    op = "promoted";
                    promotedNum.incrementAndGet();
                } else if (needEliminate(proxyInfo)) {
                    // 淘汰
                    doEliminate(proxyInfo);
                    op = "eliminate";
                    eliminateNum.incrementAndGet();
                }
                log.trace("END {}.validateAllProxy, proxy({}:{}) op={}", getClass().getSimpleName(), proxyInfo.getIp(), proxyInfo.getPort(), op);
            });

        }

        progress.setTotal(scanNum);
    }

    protected abstract int size();

    private boolean removeInvalid(ProxyInfo proxyInfo) {
        boolean ipPortOk = AddressUtils.isIpv4(proxyInfo.getIp()) && AddressUtils.isPort(proxyInfo.getPort());
        if (!ipPortOk || proxyInfo.isInvalid()) {
            Threads.execute(() -> remove(proxyInfo));
            return true;
        }
        return false;
    }

    private void doPromoted(ProxyInfo proxyInfo) {
        // 晋级
        if (ProxyRepository.SOURCE.equals(name)) {
            proxyRepositoryMap.get(USABLE).add(proxyInfo);
            remove(proxyInfo);
        } else if (ProxyRepository.USABLE.equals(name)) {
            proxyRepositoryMap.get(STABLE).add(proxyInfo);
            remove(proxyInfo);
        }
    }

    private void doEliminate(ProxyInfo proxyInfo) {
        if (ProxyRepository.SOURCE.equals(name)) {
            // 移除拉黑
            ipPortDistincter.add(proxyInfo.getIp(), proxyInfo.getPort());
            remove(proxyInfo);
        } else if (ProxyRepository.USABLE.equals(name)) {
            // 清空其历史记录并降级
            proxyInfo.cleanHistory();
            proxyRepositoryMap.get(SOURCE).add(proxyInfo);
            remove(proxyInfo);
        } else if (ProxyRepository.STABLE.equals(name)) {
            // 清空其历史记录并降级
            proxyInfo.cleanHistory();
            proxyRepositoryMap.get(USABLE).add(proxyInfo);
            remove(proxyInfo);
        }
    }

    protected boolean ableToPromoted(ProxyInfo proxyInfo) {
        // 晋级条件
        if (ProxyRepository.SOURCE.equals(name)) {
            // 能用
            return proxyInfo.getSuccessContinuityCount() > 0;
        } else if (ProxyRepository.USABLE.equals(name)) {
            // 分高
            return proxyInfo.getSuccessPercentage() > 70;
        }
        return false;
    }

    protected boolean needEliminate(ProxyInfo proxyInfo) {
        // 降级条件
        if (proxyInfo.isInvalid()) {
            return true;
        }
        if (ProxyRepository.SOURCE.equals(name)) {
            // 连续失败次数过多，且成功率拉跨
            return proxyInfo.getFailContinuityCount() > 7 && proxyInfo.getSuccessPercentage() < 10;
        } else if (ProxyRepository.USABLE.equals(name)) {
            return proxyInfo.getFailContinuityCount() > 7 && proxyInfo.getSuccessPercentage() < 20;
        } else if (ProxyRepository.STABLE.equals(name)) {
            return proxyInfo.getFailContinuityCount() > 7 && proxyInfo.getSuccessPercentage() < 30;
        }
        return false;
        // 淘汰算法：可用性差
        //return proxyInfo.getTotalCount() > 10 && (proxyInfo.getFailContinuityCount() > 4 || proxyInfo.getSuccessPercentage() < 20) ;
    }

    protected boolean validateAndUpdateProxyInfo(ProxyInfo proxyInfo) {
        AssertUtils.isTrue(AddressUtils.isIpv4(proxyInfo.getIp()) && NetUtil.isValidPort(proxyInfo.getPort()), CommonErrorCodeEnum.ILLEGAL_PARAM);
        boolean needFillInfo = ProxyRepository.USABLE.equals(name) && !infoCompleted(proxyInfo);
        ProxyTester tester = proxyTesterMap.get(needFillInfo ? ProxyTester.FILL_INFO_TESTER : ProxyTester.BASE_TESTER);

        boolean pass = tester.test(proxyInfo);
        update(proxyInfo);

        log.trace("validate proxy({}:{}) result={}", proxyInfo.getIp(), proxyInfo.getPort(), pass);
        return pass;
    }

    private boolean infoCompleted(ProxyInfo proxyInfo) {
        return StringUtils.isNoneBlank(proxyInfo.getAnonymity(), proxyInfo.getLocation(), proxyInfo.getOperator()) && proxyInfo.getSupportHttps() != null;
    }

    @Override
    public void add(ProxyInfo proxyInfo) {
        // validate
        if (AddressUtils.isIpv4(proxyInfo.getIp()) && AddressUtils.isPort(proxyInfo.getPort())) {
            if (!ipPortDistincter.contains(proxyInfo.getIp(), proxyInfo.getPort())) {
            fillBaseInfo(proxyInfo);
            doAdd(proxyInfo);
                ipPortDistincter.add(proxyInfo.getIp(), proxyInfo.getPort());
            }
        } else {
            log.info("ignore illegal proxy({},{})", proxyInfo.getIp(), proxyInfo.getPort());
        }
    }

    @Override
    public void update(ProxyInfo proxyInfo) {
        doAdd(proxyInfo);
    }

    protected abstract void doAdd(ProxyInfo proxyInfo);

    public abstract ProxyInfo remove(ProxyInfo proxyInfo);

    protected void fillBaseInfo(ProxyInfo proxyInfo) {
//        if (proxyInfo.getType() == null) {
//            proxyInfo.setType(Proxy.Type.HTTP);
//        }
        if (proxyInfo.getCreateTime() == null) {
            proxyInfo.setCreateTime(LocalDateTime.now());
        }
        if (StrUtil.isBlank(proxyInfo.getSource())) {
            proxyInfo.setSource("Unknown");
            log.warn("proxy【{}】 source is blank!", proxyInfo);
        }
    }

}